package com.imooc.rabbitmq.handack;

import com.imooc.rabbitmq.utils.ConnectionUtils;
import com.rabbitmq.client.*;

import java.io.IOException;
import java.util.concurrent.TimeoutException;

/**
 * 限流，预防消费者一上线，队列的消息全部进入消费
 */
public class Consumer {
    private static final String exchangeName = "test_ack_exchange";
    private static final String exchangeType = "topic";
    private static final String routingKey = "ack.*";
    private static final String queueName = "test_ack_queue";

    public static void main(String[] args) throws IOException, TimeoutException {
        Connection connection = ConnectionUtils.getConnection();

        //2. 通过Connection 创建一个channel
        Channel channel = connection.createChannel();

        //3.声明一个交换机   durable：是否持久化
        channel.exchangeDeclare(exchangeName,exchangeType,true,false,null);
        //声明一个队列
        channel.queueDeclare(queueName,false,false,false,null);
        //建立一个绑定关系
        channel.queueBind(queueName,exchangeName,routingKey);

        DefaultConsumer consumer = new DefaultConsumer(channel){
            @Override
            public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
                System.out.println("-------------------consume message--------------------");
                System.out.println("body:"+new String(body));
                try {
                    Thread.sleep(2000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
                if((Integer)properties.getHeaders().get("num") == 0){
                    //模拟消费失败 第0条消息消费失败，放到队列尾部，会一致消费
                    channel.basicNack(envelope.getDeliveryTag(),false,true);
                }else{
                    channel.basicAck(envelope.getDeliveryTag(),false);  //不返回ack 只接受到prefetchCount条数据
                }
            }
        };

        //监听队列   手工签收要关闭AutoAck
        channel.basicConsume(queueName,false, consumer);
    }
}
